package day3;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

public class job1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.enableCheckpointing(5000);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("bigdata1:9092")
                .setTopics("order")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setGroupId("day3job1")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        /*WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner()*/
        WatermarkStrategy<String> watermarks = WatermarkStrategy.<String>noWatermarks();

        env.fromSource(source , watermarks , "day3job1")
                        .filter(s -> s.contains("1006") || s.contains("1004"))
                                .map(s -> {
                                    String[] strings = s.split(",");
                                    return new String[]{strings[0] , strings[1] , strings[3]};
                                }).keyBy(s -> s[0])
                        .process(new ProcessFunction<String[], String[]>() {
                            ValueState<Integer> money;
                            @Override
                            public void open(Configuration parameters) throws Exception {
                                money = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("money", Integer.TYPE));
                            }
                            @Override
                            public void processElement(String[] strings, ProcessFunction<String[], String[]>.Context context, Collector<String[]> collector) throws Exception {
                                Integer value = money.value();
                                if (strings[1].equals("1004")) value += Integer.getInteger( strings[2]);
                                else value -= Integer.getInteger(strings[2]);
                                money.update(value);
                                String[] strings1 = {strings[0], String.valueOf(money.value())};
                                collector.collect(strings1);
                            }
                        }).windowAll(TumblingEventTimeWindows.of(Time.seconds(60)));
                        /*.process(new ProcessAllWindowFunction<String[], String, TimeWindow>() {
                            ValueState<String> last;
                            @Override
                            public void open(Configuration parameters) throws Exception {
                                last = getRuntimeContext().getState(new ValueStateDescriptor<String>("last", String.class));
                            }

                            @Override
                            public void process(ProcessAllWindowFunction<String[], String, TimeWindow>.Context context, Iterable<String[]> iterable, Collector<String> collector) throws Exception {
                                for (String[] strings : iterable) {
                                    strings
                                }
                            }
                        })*/

        env.execute();
    }
}
